

package com.hazelcast.jet.impl.client.protocol.task;

import com.hazelcast.client.impl.protocol.MessageTaskFactory;
import com.hazelcast.client.impl.protocol.MessageTaskFactoryProvider;
import com.hazelcast.client.impl.protocol.codec.*;
import com.hazelcast.instance.impl.Node;
import com.hazelcast.internal.util.collection.Int2ObjectHashMap;
import com.hazelcast.spi.impl.NodeEngine;

public class JetMessageTaskFactoryProvider implements MessageTaskFactoryProvider {
    private final Int2ObjectHashMap<MessageTaskFactory> factories = new Int2ObjectHashMap<>(50);
    private final Node node;

    public JetMessageTaskFactoryProvider(NodeEngine nodeEngine) {
        this.node = nodeEngine.getNode();
        initFactories();
    }

    public void initFactories() {
        factories.put(JetSubmitJobCodec.REQUEST_MESSAGE_TYPE, (cm, con) -> new JetSubmitJobMessageTask(cm, node, con));
        factories.put(JetTerminateJobCodec.REQUEST_MESSAGE_TYPE, (cm, con) -> new JetTerminateJobMessageTask(cm, node, con));
        factories.put(JetGetJobStatusCodec.REQUEST_MESSAGE_TYPE, (cm, con) -> new JetGetJobStatusMessageTask(cm, node, con));
        factories.put(JetGetJobSuspensionCauseCodec.REQUEST_MESSAGE_TYPE, (cm, con) -> new JetGetJobSuspensionCauseMessageTask(cm, node, con));
        factories.put(JetGetJobMetricsCodec.REQUEST_MESSAGE_TYPE, (cm, con) -> new JetGetJobMetricsMessageTask(cm, node, con));
        factories.put(JetGetJobIdsCodec.REQUEST_MESSAGE_TYPE, (cm, con) -> new JetGetJobIdsMessageTask(cm, node, con));
        factories.put(JetJoinSubmittedJobCodec.REQUEST_MESSAGE_TYPE, (cm, con) -> new JetJoinSubmittedJobMessageTask(cm, node, con));
        factories.put(JetGetJobSubmissionTimeCodec.REQUEST_MESSAGE_TYPE, (cm, con) -> new JetGetJobSubmissionTimeMessageTask(cm, node, con));
        factories.put(JetGetJobConfigCodec.REQUEST_MESSAGE_TYPE, (cm, con) -> new JetGetJobConfigMessageTask(cm, node, con));
        factories.put(JetUpdateJobConfigCodec.REQUEST_MESSAGE_TYPE, (cm, con) -> new JetUpdateJobConfigMessageTask(cm, node, con));
        factories.put(JetResumeJobCodec.REQUEST_MESSAGE_TYPE, (cm, con) -> new JetResumeJobMessageTask(cm, node, con));
        factories.put(JetExportSnapshotCodec.REQUEST_MESSAGE_TYPE, (cm, con) -> new JetExportSnapshotMessageTask(cm, node, con));
        factories.put(JetGetJobSummaryListCodec.REQUEST_MESSAGE_TYPE, (cm, con) -> new JetGetJobSummaryListMessageTask(cm, node, con));
        factories.put(JetGetJobAndSqlSummaryListCodec.REQUEST_MESSAGE_TYPE, (cm, con) -> new JetGetJobAndSqlSummaryListMessageTask(cm, node, con));
        factories.put(JetExistsDistributedObjectCodec.REQUEST_MESSAGE_TYPE, (cm, con) -> new JetExistsDistributedObjectMessageTask(cm, node, con));
        factories.put(JetIsJobUserCancelledCodec.REQUEST_MESSAGE_TYPE, (cm, con) -> new JetIsJobUserCancelledMessageTask(cm, node, con));
        factories.put(JetUploadJobMetaDataCodec.REQUEST_MESSAGE_TYPE, (cm, con) -> new JetUploadJobMetaDataTask(cm, node, con));
        factories.put(JetUploadJobMultipartCodec.REQUEST_MESSAGE_TYPE, (cm, con) -> new JetUploadJobMultipartTask(cm, node, con));
        factories.put(JetAddJobStatusListenerCodec.REQUEST_MESSAGE_TYPE, (cm, con) -> new JetAddJobStatusListenerMessageTask(cm, node, con));
        factories.put(JetRemoveJobStatusListenerCodec.REQUEST_MESSAGE_TYPE, (cm, con) -> new JetRemoveJobStatusListenerMessageTask(cm, node, con));
    }

    @Override
    public Int2ObjectHashMap<MessageTaskFactory> getFactories() {
        return factories;
    }
}
